package com.ryenlii.datatrans.service.impl;

import com.ryenlii.datatrans.service.IProducerService;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Service
@RequiredArgsConstructor
public class ProducerServiceImpl implements IProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private static final Logger logger = LoggerFactory.getLogger(ProducerServiceImpl.class);

    @Override
    public void sendSyncMessage(String topic, String data) throws ExecutionException, InterruptedException {
        SendResult<String, String> sendResult = kafkaTemplate.send(topic, data).get();
        RecordMetadata recordMetadata = sendResult.getRecordMetadata();
        logger.info("发送同步消息成功!发送的主题为:{}", recordMetadata.topic());
    }

    @Override
    public void sendMessage(String topic, String data) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, data);
        futureResult(future);
    }

    @Override
    public void sendMessage(ProducerRecord<String, String> record) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);
        futureResult(future);
    }

    @Override
    public void sendMessage(Message<String> message) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(message);
        futureResult(future);
    }

    @Override
    public void sendMessage(String topic, String key, String data) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, data);
        futureResult(future);
    }

    @Override
    public void sendMessage(String topic, Integer partition, String key, String data) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, partition, key, data);
        futureResult(future);
    }

    @Override
    public void sendMessage(String topic, Integer partition, Long timestamp, String key, String data) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, partition, timestamp, key, data);
        futureResult(future);
    }

    private void futureResult(CompletableFuture<SendResult<String, String>> future) {
        future.whenCompleteAsync((result, throwable) -> {
            if (throwable != null) {
                logger.error("发送消息失败!原因是：{}", throwable.getMessage());
            } else {
                RecordMetadata recordMetadata = result.getRecordMetadata();
                logger.info("发送{}消息成功!", recordMetadata.topic());
            }
        });
    }
}
